package com.uuah.server.component.model.seda;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import javax.resource.spi.work.Work;
import javax.resource.spi.work.WorkEvent;
import javax.resource.spi.work.WorkListener;

import com.uuah.api.config.IThreadingProfile;
import com.uuah.api.work.IWorkManager;
import com.uuah.config.QueueProfile;
import com.uuah.config.i18n.MessageFactory;
import com.uuah.exception.AbstractUuahException;
import com.uuah.exception.UuahRuntimeException;
import com.uuah.exception.lifecycle.InitialisationException;
import com.uuah.exception.lifecycle.LifecycleException;
import com.uuah.queue.Queue;
import com.uuah.queue.QueueSession;
import com.uuah.server.UuahEvent;
import com.uuah.server.UuahServer;
import com.uuah.server.api.IExceptionPayload;
import com.uuah.server.api.IUuahEvent;
import com.uuah.server.api.IUuahMessage;
import com.uuah.server.api.message.IUuahBusiMessage;
import com.uuah.server.api.message.IUuahBusiMessageCollection;
import com.uuah.server.api.transport.IMessageAdapter;
import com.uuah.server.component.service.AbstractService;
import com.uuah.server.config.i18n.ServerMessages;
import com.uuah.server.exception.AbstractUuahServerException;
import com.uuah.server.exception.FailedToQueueEventException;
import com.uuah.server.exception.MessagingException;
import com.uuah.server.exception.service.ServiceException;
import com.uuah.server.message.ExceptionPayload;
import com.uuah.server.message.UuahMessage;
import com.uuah.server.transport.NullPayload;
import com.uuah.server.transport.RequestContextHelper;

public class SedaService extends AbstractService implements Work, WorkListener {
	private static final long serialVersionUID = 7711976708670893015L;

	private static final String QUEUE_NAME_SUFFIX = ".service";

	protected IWorkManager workManager;

	protected Integer queueTimeout;

	protected IThreadingProfile threadingProfile;

	protected QueueProfile queueProfile;

	protected Queue queue;

	public SedaService() {
		super();
	}

	protected synchronized void doInitialise() throws InitialisationException {
		if (threadingProfile == null) {
			threadingProfile = uuahContext.getDefaultServiceThreadingProfile();
		}
		workManager = threadingProfile.createWorkManager(getName());

		if (queueProfile == null) {
			queueProfile = ((SedaModel) model).getQueueProfile();
		}

		if (queueTimeout == null) {
			setQueueTimeout(new Integer(((SedaModel) model).getQueueTimeout()));
		}

		try {
			if (name == null) {
				throw new InitialisationException(
						MessageFactory
								.createStaticMessage("Service has no name to identify it"),
						this);
			}
			// ���ö���
			queueProfile.configureQueue(getQueueName(), uuahContext
					.getQueueManager());
			queue = uuahContext.getQueueManager().getQueueSession().getQueue(
					getQueueName());
			if (queue == null) {
				throw new InitialisationException(MessageFactory
						.createStaticMessage("Queue not created for service "
								+ name), this);
			}
		} catch (InitialisationException e) {
			throw e;
		} catch (Throwable e) {
			throw new InitialisationException(ServerMessages
					.objectFailedToInitialise("Service Queue"), e, this);
		}
	}

	protected void doForceStop() throws AbstractUuahServerException {
		doStop();
	}

	protected void doStop() throws AbstractUuahServerException {
		if (queue != null && queue.size() > 0) {
			try {
				stopping.whenFalse(null);
			} catch (InterruptedException e) {
			}
		}
		workManager.dispose();
	}

	protected void doStart() throws AbstractUuahException {
		try {
			workManager.start();
			workManager.scheduleWork(this, IWorkManager.INDEFINITE, null, this);
		} catch (Exception e) {
			throw new LifecycleException(ServerMessages
					.failedToStart("Service: " + name), e, this);
		}
	}

	protected void doDispose() {
		queue = null;
		if (workManager != null) {
			workManager.dispose();
		}
	}

	protected void doDispatch(IUuahEvent event)
			throws AbstractUuahServerException {
		if (logger.isDebugEnabled()) {
			logger.debug("Service: " + name
					+ " has received asynchronous event on: "
					+ event.getEndpoint().getEndpointURI());
		}

		try {
			enqueue(event);
		} catch (Exception e) {
			FailedToQueueEventException e1 = new FailedToQueueEventException(
					ServerMessages.interruptedQueuingEventFor(this.getName()),
					event.getMessage(), this, e);
			handleException(e1);
		}

		if (logger.isTraceEnabled()) {
			logger.trace("UuahEvent added to queue for: " + name);
		}
	}

	protected IUuahMessage doSend(IUuahEvent event)
			throws AbstractUuahServerException {
		IUuahMessage result = null;
		try {
			if (logger.isDebugEnabled()) {
				logger.debug(this + " : got proxy for " + event.getId() + " = "
						+ component);
			}
			IUuahBusiMessageCollection collection = (IUuahBusiMessageCollection) event
					.getMessage().getPayload();
			List responseCollection = new CopyOnWriteArrayList();
			for (IUuahBusiMessage uuahBusiMessage : collection
					.getMessagesAsArray()) {
				IMessageAdapter adapter = event.getEndpoint().getConnector()
						.getMessageAdapter(uuahBusiMessage);
				IUuahMessage uuahMessage = new UuahMessage(adapter);
				uuahMessage.applyTransformer(event.getEndpoint().getConnector()
						.getDefaultRequestTransformer());
				IUuahEvent uuahEvent = new UuahEvent(uuahMessage, event
						.getEndpoint(), event.getSession(), event
						.isSynchronous());
				result = invokeComponent(uuahEvent);

				responseCollection.add(result.getPayload());
			}
			result.setPayload(responseCollection);
		} catch (Exception e) {
			if (e instanceof MessagingException) {
				handleException(e);
			} else {
				handleException(new MessagingException(ServerMessages
						.eventProcessingFailedFor(getName()), event
						.getMessage(), e));
			}
			if (result == null) {
				result = new UuahMessage(NullPayload.getInstance(),
						RequestContextHelper.getEvent().getMessage());
			}
			IExceptionPayload exceptionPayload = result.getExceptionPayload();
			if (exceptionPayload == null) {
				exceptionPayload = new ExceptionPayload(e);
			}
			result.setExceptionPayload(exceptionPayload);
		}
		return result;
	}

	public int getQueueSize() {
		if (queue == null) {
			logger.warn(new InitialisationException(MessageFactory
					.createStaticMessage("Queue not created for service "
							+ name), this).getMessage());
			return -1;
		}
		return queue.size();
	}

	private String getQueueName() {
		return name + QUEUE_NAME_SUFFIX;
	}

	public void run() {
		UuahEvent event = null;
		QueueSession queueSession = uuahContext.getQueueManager()
				.getQueueSession();

		while (!stopped.get()) {
			try {
				if (paused.get()) {
					paused.whenFalse(null);

					if (stopping.get()) {
						if (!queueProfile.isPersistent()
								&& (queueSession != null && getQueueSize() > 0)) {
							logger
									.warn(ServerMessages
											.stopPausedSedaServiceNonPeristentQueueMessageLoss(
													getQueueSize(), this)
											.getMessage());
						}
						stopping.set(false);
						break;
					}
				}

				if (stopping.get()) {
					if (queueProfile.isPersistent()
							|| (queueSession == null || getQueueSize() <= 0)) {
						stopping.set(false);
						break;
					}
				}

				event = (UuahEvent) dequeue();
				if (event != null) {
					if (logger.isDebugEnabled()) {
						logger.debug("Service: " + name
								+ " dequeued event on: "
								+ event.getEndpoint().getEndpointURI());
					}
					workManager.scheduleWork(new ComponentStageWorker(event),
							IWorkManager.INDEFINITE, null, this);
				}
			} catch (Exception e) {
				if (e instanceof InterruptedException) {
					stopping.set(false);
					break;
				}
				if (e instanceof AbstractUuahServerException) {
					handleException(e);
				} else {
					handleException(new ServiceException(ServerMessages
							.eventProcessingFailedFor(name),
							(event == null ? null : event.getMessage()), this,
							e));
				}
			}
		}
	}

	public void release() {
		stopping.set(false);
	}

	protected void enqueue(IUuahEvent event) throws Exception {
		if (queue == null) {
			throw new InitialisationException(MessageFactory
					.createStaticMessage("Queue not created for service "
							+ name), this);
		}
		if (logger.isDebugEnabled()) {
			logger.debug("Service " + name + " putting event on queue "
					+ queue.getName() + ": " + event);
		}
		queue.put(event);
		UuahServer.getServerStats().incQueuedEvent();
	}

	protected IUuahEvent dequeue() throws Exception {
		if (queue == null) {
			throw new InitialisationException(MessageFactory
					.createStaticMessage("Queue not created for service "
							+ name), this);
		}

		if (getQueueTimeout() == null) {
			throw new InitialisationException(ServerMessages
					.noServiceQueueTimeoutSet(this), this);
		} else {
			UuahServer.getServerStats().decQueuedEvent();
			return (IUuahEvent) queue.poll(getQueueTimeout().intValue());
		}
	}

	public void workAccepted(WorkEvent event) {
		handleWorkException(event, "workAccepted");
	}

	public void workRejected(WorkEvent event) {
		handleWorkException(event, "workRejected");
	}

	public void workStarted(WorkEvent event) {
		handleWorkException(event, "workStarted");
	}

	public void workCompleted(WorkEvent event) {
		handleWorkException(event, "workCompleted");
	}

	protected void handleWorkException(WorkEvent event, String type) {
		Throwable e;

		if (event != null && event.getException() != null) {
			e = event.getException();
		} else {
			return;
		}

		if (event.getException().getCause() != null) {
			e = event.getException().getCause();
		}

		logger.error("Work caused exception on '" + type
				+ "'. Work being executed was: " + event.getWork().toString());

		if (e instanceof Exception) {
			handleException((Exception) e);
		} else {
			throw new UuahRuntimeException(ServerMessages
					.componentCausedErrorIs(this.getName()), e);
		}
	}

	public Object getInstance() throws AbstractUuahServerException {
		throw new UnsupportedOperationException(
				"Direct access to underlying service object is not allowed in the SedaModel.  If this is for a unit test, make sure you are using the TestSedaModel ('seda-test')");
	}

	public QueueProfile getQueueProfile() {
		return queueProfile;
	}

	public void setQueueProfile(QueueProfile queueProfile) {
		this.queueProfile = queueProfile;
	}

	public Integer getQueueTimeout() {
		return queueTimeout;
	}

	public void setQueueTimeout(Integer queueTimeout) {
		this.queueTimeout = queueTimeout;
	}

	public IThreadingProfile getThreadingProfile() {
		return threadingProfile;
	}

	public void setThreadingProfile(IThreadingProfile threadingProfile) {
		this.threadingProfile = threadingProfile;
	}

	public IWorkManager getWorkManager() {
		return workManager;
	}

	public void setWorkManager(IWorkManager workManager) {
		this.workManager = workManager;
	}

	private class ComponentStageWorker implements Work {
		private IUuahEvent event;

		public ComponentStageWorker(IUuahEvent event) {
			this.event = event;
		}

		public void run() {
			try {
				event = RequestContextHelper.criticalSetEvent(event);
				IUuahMessage result = null;
				IUuahBusiMessageCollection collection = (IUuahBusiMessageCollection) event
						.getMessage().getPayload();
				List responseCollection = new CopyOnWriteArrayList();
				for (IUuahBusiMessage uuahBusiMessage : collection
						.getMessagesAsArray()) {
					IMessageAdapter adapter = event.getEndpoint()
							.getConnector().getMessageAdapter(uuahBusiMessage);
					IUuahMessage uuahMessage = new UuahMessage(adapter);

					uuahMessage.applyTransformer(event.getEndpoint()
							.getConnector().getDefaultRequestTransformer());
					IUuahEvent uuahEvent = new UuahEvent(uuahMessage, event
							.getEndpoint(), event.getSession(), event
							.isSynchronous());

					result = invokeComponent(uuahEvent);
					responseCollection.add(result.getPayload());
				}
				result.setPayload(responseCollection);
				processAsyncResponse(result, event);
			} catch (Exception e) {
				if (e instanceof MessagingException) {
					handleException(e);
				} else {
					handleException(new MessagingException(ServerMessages
							.eventProcessingFailedFor(getName()), event
							.getMessage(), e));
				}
			}
		}

		public void release() {
			// no-op
		}
	}
}
